package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TrafficPageViewBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficScFlow {
    public static void main(String[] args) throws Exception {
        //TODO 1 获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 设置检查点和状态后端
    /*
    env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
    env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
    env.setStateBackend(new HashMapStateBackend());
    env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
    System.setProperty("HADOOP_USER_NAME", "atguigu");
     */

        //TODO 3 读取对应主题数据
        String page_topic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujTopic = "dwd_traffic_user_jump_detail";
        String groupID = "dws_traffic_sc_flow";

        DataStreamSource<String> pageStream = env.addSource(KafkaUtil.getKafkaConsumer(page_topic, groupID));
        DataStreamSource<String> uvStream = env.addSource(KafkaUtil.getKafkaConsumer(uvTopic, groupID));
        DataStreamSource<String> ujStream = env.addSource(KafkaUtil.getKafkaConsumer(ujTopic, groupID));

        //TODO 4 转化数据格式
        SingleOutputStreamOperator<TrafficPageViewBean> pageBeanStream = pageStream.map(new MapFunction<String,
                TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String lastPageId = jsonObject.getString("last_page_id");
                String duringTime = common.getString("during_time");
                long sv = 0;
                if (lastPageId == null) {
                    sv = 1;
                }
                return TrafficPageViewBean.builder()
                        .vc(common.getString("vc"))
                        .sc(common.getString("sc"))
                        .svCt(sv)
                        .pvCt(1L)
                        .uvCt(0L)
                        .ujCt(0L)
                        .durSum(Long.parseLong(duringTime==null? "0":duringTime))
                        .ts(jsonObject.getLong("ts"))
                        .build();
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> uvBeanStream = uvStream.map(new MapFunction<String,
                TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return TrafficPageViewBean.builder()
                        .vc(common.getString("vc"))
                        .sc(common.getString("sc"))
                        .svCt(0L)
                        .pvCt(0L)
                        .uvCt(1L)
                        .durSum(0L)
                        .ujCt(0L)
                        .ts(jsonObject.getLong("ts"))
                        .build();
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> ujBeanStream = ujStream.map(new MapFunction<String,
                TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return TrafficPageViewBean.builder()
                        .vc(common.getString("vc"))
                        .sc(common.getString("sc"))
                        .uvCt(0L)
                        .svCt(0L)
                        .pvCt(0L)
                        .ujCt(1L)
                        .durSum(0L)
                        .ts(jsonObject.getLong("ts"))
                        .build();
            }
        });

        //TODO 5 三个表合并
        DataStream<TrafficPageViewBean> unionStream = pageBeanStream.union(uvBeanStream).union(ujBeanStream);

        //TODO 6 设置水位线
        SingleOutputStreamOperator<TrafficPageViewBean> beanWithWatermarkStream =
                unionStream.assignTimestampsAndWatermarks(WatermarkStrategy.
                        <TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));
        unionStream.print("union>>");

        //TODO 7 根据来源分组后并计算
        SingleOutputStreamOperator<TrafficPageViewBean> resultStream = beanWithWatermarkStream.keyBy(s -> s.getSc())
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .allowedLateness(Time.seconds(2))
                .apply( new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TrafficPageViewBean> input,
                                      Collector<TrafficPageViewBean> out) throws Exception {

                        long svCt = 0;
                        long pvCt = 0;
                        long durSum = 0;
                        long ujCt = 0;
                        for (TrafficPageViewBean trafficPageViewBean : input) {
                            svCt += trafficPageViewBean.getSvCt();
                            pvCt += trafficPageViewBean.getPvCt();
                            durSum += trafficPageViewBean.getDurSum();
                            ujCt += trafficPageViewBean.getUjCt();
                        }
                        if (svCt != 0 && svCt != 0 && svCt != 0) {
                            long avgSessViPage = pvCt / svCt;
                            long avgSeDurt = durSum / svCt;
                            long bounRate = ujCt / svCt;


                            TrafficPageViewBean trafficPageViewBean = input.iterator().next();
                            trafficPageViewBean.setSvCt(svCt);
                            trafficPageViewBean.setPvCt(pvCt);
                            trafficPageViewBean.setDurSum(durSum);
                            trafficPageViewBean.setUjCt(ujCt);
                            trafficPageViewBean.setAvgSessViPage(avgSessViPage);
                            trafficPageViewBean.setAvgSeDurt(avgSeDurt);
                            trafficPageViewBean.setBounRate(bounRate);
                            trafficPageViewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            trafficPageViewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            trafficPageViewBean.setTs(System.currentTimeMillis());
                        }
                    }
                })
             ;
        resultStream.print("result>>");

                //TODO 8 写入clickhouse中
        resultStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_traffic_sc_flow values(?,?,?," +
                "?,?,?,?,?," +
                "?,?,?,?,?) "));
        // TODO  执行任务
        JobExecutionResult execute = env.execute();
    }

}
